package com.gitee.baqkft.code.ui.dlg;

import com.gitee.baqkft.code.ui.layout.GBC;
import com.gitee.baqkft.code.ui.utils.ConfigFactory;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.swing.*;
import java.awt.*;
import java.awt.event.ActionEvent;
import java.util.Arrays;

/**
 * @Author: chenmingcai
 * @Date 2023/3/3 09:40
 **/

public class KafkaToolsFrame extends JFrame
{
    /**
     * 服务地址
     */
    private JTextField server;

    /**
     * 接收主题
     */
    private JTextField receiveTopics;

    /**
     * 消费组
     */
    private JTextField receiveGroupId;

    /**
     * 接收到的消息
     */
    private JTextArea receivedMessages;

    /**
     * 发送主题
     */
    private JTextField sendTopics;

    /**
     * 发送消息
     */
    private JTextArea messages;

    /**
     * 生产者
     */
    private Producer<String, String> producer;

    /**
     * 消费者
     */
    private Consumer<String, String> consumer;

    public KafkaToolsFrame() throws HeadlessException
    {
        setDefaultCloseOperation(JFrame.DISPOSE_ON_CLOSE);
        setTitle("Kafka消息助手");

        Toolkit kit = Toolkit.getDefaultToolkit();
        Dimension screenSize = kit.getScreenSize();
        int width = screenSize.width / 2;
        int height = screenSize.height / 2;
        setSize(width, height);

        //初始化界面
        this.initUI();
    }

    /**
     * 初始化界面
     */
    private final void initUI()
    {
        setLayout(new GridBagLayout());
        GBC gbc = new GBC(0, 0, 1, 1).setWeight(1, 0)
                .setFill(GridBagConstraints.HORIZONTAL);
        add(initHeaderUI(), gbc);

        gbc = new GBC(0, 1, 1, 1).setWeight(1, 1).setFill(GBC.BOTH);
        add(initReceiveMessageUI(), gbc);

        gbc = new GBC(0, 2, 1, 1)
                .setWeight(1, 1)
                .setFill(GBC.BOTH);
        add(initSendMessageUI(), gbc);
    }

    /**
     * 顶部界面
     * @return
     */
    private final JPanel initHeaderUI()
    {
        JPanel panel = new JPanel();
        panel.setLayout(new GridBagLayout());

        JLabel label = new JLabel("bootstrap-server:");
        GBC gbc = new GBC(0, 0, 1, 1)
                .setWeight(0, 0).setFill(GBC.NONE);
        panel.add(label, gbc);

        server = new JTextField("localhost:9092");
        gbc = new GBC(1, 0, 1, 1)
                .setWeight(1, 0).setFill(GBC.HORIZONTAL);
        panel.add(server, gbc);

        JButton connect = new JButton("连接");
        gbc = new GBC(3, 0, 1, 1)
                .setWeight(0, 0).setFill(GBC.NONE);
        panel.add(connect, gbc);
        connect.addActionListener(this::toConnect);

        return panel;
    }

    /**
     * 消息接收界面
     * @return
     */
    private final JPanel initReceiveMessageUI()
    {
        JPanel panel = new JPanel();
        panel.setLayout(new GridBagLayout());
        panel.setBorder(BorderFactory.createTitledBorder("Receive Message"));


        JLabel label = new JLabel("topic");
        GBC gbc = new GBC(0, 0, 1, 1)
                .setWeight(0, 0).setFill(GBC.NONE);
        panel.add(label, gbc);

        receiveTopics = new JTextField();
        gbc = new GBC(1, 0, 1, 1)
                .setWeight(1, 0)
                .setFill(GBC.HORIZONTAL);
        panel.add(receiveTopics, gbc);

        label = new JLabel("group.id");
        gbc = new GBC(0, 1, 1, 1)
                .setWeight(0, 0).setFill(GBC.NONE);
        panel.add(label, gbc);

        receiveGroupId = new JTextField();
        gbc = new GBC(1, 1, 1, 1)
                .setWeight(1, 0)
                .setFill(GBC.HORIZONTAL);
        panel.add(receiveGroupId, gbc);
        JButton button = new JButton("订阅");
        gbc = new GBC(2, 1, 1, 1)
                .setFill(GBC.NONE);
        panel.add(button, gbc);
        button.addActionListener(this::toSubscribe);

        receivedMessages = new JTextArea();
        receivedMessages.setLineWrap(true);
        receivedMessages.setWrapStyleWord(true);
        gbc = new GBC(0, 2, 3, 1).setWeight(1, 1)
                .setFill(GBC.BOTH);
        JScrollPane msgArea = new JScrollPane();
        msgArea.setViewportView(receivedMessages);
        panel.add(msgArea, gbc);

        return panel;
    }

    /**
     * 消息发送界面
     * @return
     */
    private final JPanel initSendMessageUI()
    {
        JPanel panel = new JPanel();
        panel.setLayout(new GridBagLayout());
        panel.setBorder(BorderFactory.createTitledBorder("send Message"));

        JLabel label = new JLabel("topic");
        GBC gbc = new GBC(0, 0, 1, 1)
                .setWeight(0, 0)
                .setFill(GBC.NONE);
        panel.add(label, gbc);

        sendTopics = new JTextField();
        gbc = new GBC(1, 0, 1, 1)
                .setWeight(1, 0)
                .setFill(GBC.HORIZONTAL);
        panel.add(sendTopics, gbc);

        JScrollPane msgArea = new JScrollPane();
        messages = new JTextArea();
        messages.setLineWrap(true);
        messages.setWrapStyleWord(true);
        msgArea.setViewportView(messages);
        gbc = new GBC(0, 1, 2, 1)
                .setWeight(1, 1)
                .setFill(GBC.BOTH);
        panel.add(msgArea, gbc);

        JButton send = new JButton("发送");
        gbc = new GBC(0, 2).setFill(GBC.NONE);
        panel.add(send, gbc);

        send.addActionListener(this::toSendMessage);

        return panel;
    }

    /**
     * 连接
     * @param event
     */
    private void toConnect(ActionEvent event)
    {
        JButton button = (JButton) event.getSource();
        if("连接".equals(button.getText()))
        {
            //执行连接操作
            String address = server.getText().trim();
            if("".equals(address))
            {
                JOptionPane.showConfirmDialog(this, "服务器地址必填", "提示", JOptionPane.YES_OPTION);
                return;
            }

            if(producer != null)
                producer.close();

            try
            {
                producer = new KafkaProducer<>(ConfigFactory.getKafkaProducerConfig(address));
                button.setText("断开");
            }
            catch (Exception ex)
            {
                JOptionPane.showMessageDialog(this, "连接失败: " + ex.getMessage());
            }
        }
        else
        {
            if(producer != null)
                producer.close();
            button.setText("连接");
            producer = null;
        }
    }

    /**
     * 发送消息
     * @param event
     */
    private void toSendMessage(ActionEvent event)
    {
        if(producer == null)
        {
            JOptionPane.showMessageDialog(this, "请先连接服务器");
            return;
        }

        String topics = sendTopics.getText().trim();
        String msg = messages.getText().trim();
        if("".equals(topics) || "".equals(msg))
        {
            return;
        }

        producer.send(new ProducerRecord<>(topics, msg));
    }

    /**
     * 订阅消息, 多个topic用逗号隔开
     * @param event
     */
    private void toSubscribe(ActionEvent event)
    {
        JButton button = (JButton) event.getSource();
        if("订阅".equals(button.getText()))
        {
            if(consumer != null)
                consumer.close();

            String address = server.getText().trim();
            if("".equals(address))
            {
                JOptionPane.showConfirmDialog(this, "服务器地址必填", "提示", JOptionPane.YES_OPTION);
                return;
            }
            String groupid = receiveGroupId.getText().trim();
            if("".equals(groupid))
            {
                JOptionPane.showMessageDialog(this, "group.id必填");
                return;
            }

            String topics = receiveTopics.getText().trim();

            try
            {
                consumer = new KafkaConsumer<>(ConfigFactory.getKafkaConsumerConfig(address, groupid));
                consumer.subscribe(Arrays.asList(topics.split(",")));

                //订阅
                button.setText("取消");
            }
            catch (Exception ex)
            {
                JOptionPane.showMessageDialog(this, "订阅异常: " + ex.getMessage());
            }
        }
        else
        {
            button.setText("订阅");
        }
    }

    @Override
    public void dispose()
    {
        if(producer != null)
            producer.close();

        if(consumer != null)
            consumer.close();

        super.dispose();
    }
}
